package aggregation.system.peer.aggregation;

import aggregation.simulator.snapshot.Snapshot;
import common.configuration.AggregationConfiguration;
import common.peer.PeerAddress;
import java.util.ArrayList;

import cyclon.system.peer.cyclon.CyclonPartnersPort;
import cyclon.system.peer.cyclon.CyclonPartnersRequest;
import cyclon.system.peer.cyclon.CyclonPartnersResponse;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Positive;
import se.sics.kompics.network.Network;
import se.sics.kompics.timer.SchedulePeriodicTimeout;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timeout;
import se.sics.kompics.timer.Timer;

public final class Aggregation extends ComponentDefinition {

    Positive<CyclonPartnersPort> partnersPort = positive(CyclonPartnersPort.class);
    Positive<Network> networkPort = positive(Network.class);
    Positive<Timer> timerPort = positive(Timer.class);
    private PeerAddress self;
    private long period;
    //private long periodAgg;
    //num: my bandwidth
    private double num;
    //avgBw: estimation of the average bandwidth
    private double avgBw;
    //nums: list of num of each node
    /*estimation 1 and 2*/
    private Map<PeerAddress, Double> estimations;
    private Map<PeerAddress, Double> bandwidths;
    /*estimation 3*/
    private double[] estimationsWindow;
    private double[] bandwidthsWindow;
    private int windowSize;
    private int i;
    //private double bwEstimation;
    private ArrayList<PeerAddress> cyclonPartners;
    private AggregationConfiguration aggregationConfiguration;

//-------------------------------------------------------------------	
    public Aggregation() {
        cyclonPartners = new ArrayList<PeerAddress>();

        subscribe(handleInit, control);
        subscribe(handleRequestPartners, timerPort);
        subscribe(handleRequestNumRequest, timerPort);
        subscribe(handleAggregationTimer, timerPort);
        subscribe(handleRecvPartners, partnersPort);
        subscribe(handleRcvNumRequest, networkPort);
        subscribe(handleRcvNumResponse, networkPort);


    }
//-------------------------------------------------------------------	
    Handler<AggregationInit> handleInit = new Handler<AggregationInit>() {

        public void handle(AggregationInit init) {
            self = init.getSelf();
            num = init.getNum();
            aggregationConfiguration = init.getConfiguration();
            period = aggregationConfiguration.getPeriod();
            avgBw = num;
            windowSize = 10;
            /*estimation 1 and 2*/
            estimations = new HashMap<PeerAddress, Double>();
            bandwidths = new HashMap<PeerAddress, Double>();


            /*estimation 3*/
            estimationsWindow = new double[windowSize];
            bandwidthsWindow = new double[windowSize];
            i = 0;

            for (int j = 0; j < windowSize; j++) {
                estimationsWindow[j] = num;
                bandwidthsWindow[j] = num;
            }



            SchedulePeriodicTimeout rst = new SchedulePeriodicTimeout(period, period);
            rst.setTimeoutEvent(new AggregationSchedule(rst));
            trigger(rst, timerPort);


            ScheduleTimeout t = new ScheduleTimeout(10000L); //???
            Timeout te = new AggregationTimer(t);
            t.setTimeoutEvent(te);
            trigger(t, timerPort);



            Snapshot.updateNum(self, num);
            Snapshot.updateEstimatedBw(self, avgBw);


            CyclonPartnersRequest request = new CyclonPartnersRequest();
            trigger(request, partnersPort);
        }
    };
//-------------------------------------------------------------------	
    Handler<AggregationSchedule> handleRequestPartners = new Handler<AggregationSchedule>() {

        public void handle(AggregationSchedule event) {
            CyclonPartnersRequest request = new CyclonPartnersRequest();
            trigger(request, partnersPort);

            //Snapshot.updateNum(self, num);

        }
    };
//-------------------------------------------------------------------	
    Handler<CyclonPartnersResponse> handleRecvPartners = new Handler<CyclonPartnersResponse>() {

        public void handle(CyclonPartnersResponse event) {
            cyclonPartners = event.getPartners();
            Snapshot.updateCyclonPartners(self, cyclonPartners);
            //requestNumRequest();
        }
    };
    Handler<AggregationRequestSchedule> handleRequestNumRequest = new Handler<AggregationRequestSchedule>() {

        public void handle(AggregationRequestSchedule event) {
            //System.err.println("Peer "+self+ " timer aggregation request");
            requestNumRequest();


        }
    };
//-------------------------------------------------------------------
// In this method a node receive the request of Num and answer with
// the bandwidth estimated
//-------------------------------------------------------------------
    Handler<NumRequest> handleRcvNumRequest = new Handler<NumRequest>() {

        @Override
        public void handle(NumRequest event) {
            //cyclonPartners.add(event.getMSPeerSource());

            //System.err.println("### Peer " + self + ": received NumRequest from " + event.getMSPeerSource());
            trigger(new NumResponse(self, event.getMSPeerSource(), num, avgBw), networkPort);
            updateEstimatedAvgBw(event);
        }
    };
//-------------------------------------------------------------------
// In this method a node receive the response of Num and update
// the bandwidth estimated
//-------------------------------------------------------------------
    Handler<NumResponse> handleRcvNumResponse = new Handler<NumResponse>() {

        @Override
        public void handle(NumResponse event) {
            //System.err.println("--- Peer " + self + ": received response from " + event.getMSPeerSource() + " avgBw= " + event.getEstimatedBw());

            //System.err.println("--- Peer " + self + ": previous averageBw= " + avgBw);

            updateEstimatedAvgBw(event);
            
            //System.err.println("--- Peer " + self + ": new averageBw= " + avgBw);

            Snapshot.updateEstimatedBw(self, avgBw);
        }
    };
    Handler<AggregationTimer> handleAggregationTimer = new Handler<AggregationTimer>() {

        @Override
        public void handle(AggregationTimer e) {
            SchedulePeriodicTimeout rst2 = new SchedulePeriodicTimeout(period, period);
            rst2.setTimeoutEvent(new AggregationRequestSchedule(rst2));
            trigger(rst2, timerPort);
        }
    };

    private void requestNumRequest() {
        //System.err.println("### Peer " + self + " requests Num to random node");

        if (cyclonPartners.isEmpty()) {

            return;
        }
        Random random = new Random();
        PeerAddress randomDest = cyclonPartners.get(random.nextInt(cyclonPartners.size()));
        //System.err.println("### Peer " + self + " requests Num to "+randomDest);

        if (randomDest != null) {
            trigger(new NumRequest(self, randomDest, num, avgBw), networkPort);
            //System.err.println("### Peer " + self + " requested Num to " + randomDest);
        }

    }

    private void updateEstimatedAvgBw(NumMessage event){
        updateEstimatedAvgBw1(event);
    }
    
    private void updateEstimatedAvgBw1(NumMessage event) {

        estimations.put(event.getMSPeerSource(), event.getEstimatedBw());
        bandwidths.put(event.getMSPeerSource(), event.getNum());
        avgBw = num;
        for (PeerAddress peer : estimations.keySet()) {
            avgBw += bandwidths.get(peer);
            //System.err.println("--- Peer " + self + ": estimation of = " + peer + ": "+estimations.get(peer));
        }
        avgBw /= (bandwidths.size() + 1);

    }

    private void updateEstimatedAvgBw2(NumMessage event) {

        estimations.put(event.getMSPeerSource(), event.getEstimatedBw());
        bandwidths.put(event.getMSPeerSource(), event.getNum());


        avgBw += event.getEstimatedBw();
        avgBw /= 2;

    }

    private void updateEstimatedAvgBw3(NumMessage event) {

        estimationsWindow[i % windowSize] = event.getEstimatedBw();
        bandwidthsWindow[i % windowSize] = event.getNum();
        i++;
        avgBw = 0;
        for (int j = 0; j < windowSize; j++) {
            avgBw += estimationsWindow[j];
        }
        avgBw /= (windowSize);

    }
}
